package com.bgq.trxauth.listener;


import com.alibaba.fastjson2.JSONObject;
import com.bgq.trxauth.mapper.AuthMapper;
import com.bgq.trxauth.model.Auth;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;

@Component
public class MessageListener {

    @Autowired
    private AuthMapper authMapper;

    @Autowired
    private KafkaTemplate kafkaTemplate;

    @KafkaListener(topics = {"cat"}, id = "bgq-1")
    public void onMessage(ConsumerRecord<?, ?> record, Acknowledgment acknowledgment) {
        // 消费的哪个topic、partition的消息,打印出消息内容
        String msg = record.topic() + "-" + record.partition() + "-" + record.value();
        System.out.println(msg);

        Object value = record.value();
        JSONObject jsonData = JSONObject.parseObject(value.toString());


        String authCode = jsonData.getString("authCode");
        String userId = jsonData.getString("userId");
        String payId = jsonData.getString("payId");
        String msgId = jsonData.getString("msgId");

        // add auth
        Auth auth = new Auth();
        long authId = new Random().nextLong();
        if (authId < 0) {
            authId = authId * -1;
        }
        auth.setId(authId);
        auth.setUserId(userId);
        auth.setPayRecordRefId(payId);
        auth.setAuthCode(authCode);
        auth.setCreateTime(new Date());
        auth.setUpdateTime(new Date());
        try {
            authMapper.insert(auth);
            // success

            // send success message to kafka
            Map<String, Object> successMap = new HashMap<>();
            successMap.put("msgId", msgId);
            kafkaTemplate.send("fox", JSONObject.from(successMap).toString());

        } catch (Exception exception) {
            exception.printStackTrace();
            // failure
        }

        // 如果开启手动确认的话，那需要 在配置文件中 配置  listener  的设置,ack-mode
        acknowledgment.acknowledge();
    }

}
